#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import locale
import shlex
import subprocess
from typing import Callable, Any, Optional, Union

from pydantic import parse_obj_as

from common.common import is_aix
from common.exception.common_exception import ErrCodeException
from oracle import logger
from oracle.common.common import login_oracle_database_with_params, exec_sql_cmd, parse_html_result, \
    get_db_user_and_db_password, close_child
from oracle.common.constants import PexpectResult, ErrorCode, TIME_FORMAT, ListTableByOwnerParam
from oracle.common.sql_utils import split_sql_statements
from oracle.common.user_env_common import build_user_env
from oracle.schemas.oracle_schemas import DBInfo, InstanceInfo, PDBInfo, SCNAndTime, DBFile

SELECT_SETTING = ["set linesize 32767;", "set pagesize 9999;", "set timing off;", "set heading on;"]


def exec_select_cmd(
        params: dict,
        sql_cmd: str,
        parse_func: Optional[Callable[[dict], Any]] = None,
        timeout: int = 30
) -> Union[list[dict], list[Any]]:
    if is_aix():
        return exec_aix_select_cmd(params, sql_cmd, parse_func, timeout)
    logger.info(f"exec oracle sqlcmd, sql_cmd {sql_cmd}")
    pid = params.get("pid")
    result, child = login_oracle_database_with_params(pid, params, timeout)
    if not result:
        logger.info(f"login database error,pid:{pid}")
        raise ErrCodeException(ErrorCode.ORACLE_EXEC_SQL_FAILED.value, [sql_cmd], "Database connection failed")
    try:
        for setting in SELECT_SETTING:
            child.sendline(setting)
        return_code, std_out = exec_sql_cmd(child, sql_cmd, PexpectResult.FILE_PATH)
        logger.debug(f"return_code:{return_code}, std_out:{std_out}")
        if not return_code:
            logger.info(f"query oracle sqlcmd:{sql_cmd} error,{std_out}")
            return []
    except Exception as exception:
        logger.error(f'exec cmd {sql_cmd} failed. {str(exception)}')
        raise ErrCodeException(ErrorCode.ORACLE_EXEC_SQL_FAILED.value, [sql_cmd], "execute sql error")
    finally:
        close_child(child)
    if PexpectResult.FILE_PATH[3] in str(std_out):
        logger.info(f"query result is empty:{sql_cmd}")
        return []
    result = parse_html_result(std_out)
    if len(result) == 0:
        return []
    logger.info(f"result:{result},pid:{pid}")
    if not parse_func:
        return result
    return [parse_func(db) for db in result]


def exec_aix_select_cmd(
        params: dict,
        sql_cmd: str,
        parse_func: Optional[Callable[[dict], Any]] = None,
        timeout: int = 30
) -> Union[list[dict], list[Any]]:
    logger.info(f"exec aix oracle sqlcmd, sql_cmd {sql_cmd}")
    pid = params.get("pid")
    instance_name = params.get("instance_name", "")
    db_install_user = params.get("db_install_user", "")
    oracle_user_env = build_user_env(db_install_user, inst_name=instance_name)
    exec_cmd = f"su - {db_install_user} -c '{oracle_user_env} sqlplus -M \"HTML ON\" /nolog'"
    logger.info(f"exec_cmd:{exec_cmd}")
    try:
        proc = subprocess.Popen(shlex.split(exec_cmd), stdout=subprocess.PIPE, encoding=locale.getdefaultlocale()[1],
                                stderr=subprocess.PIPE, stdin=subprocess.PIPE, errors='ignore', universal_newlines=True,
                                bufsize=1)
        db_user = params.get("db_user", "")
        db_password = params.get("db_password", "")
        if not db_password:
            db_user, db_password = get_db_user_and_db_password(params, pid)
        inputs = []
        db_password = f'"{db_password}"'
        if db_user:
            inputs.append(f'connect {db_user}/{db_password} as sysdba')
        else:
            inputs.append(f'connect sys/'' as sysdba')
        inputs += SELECT_SETTING
        inputs += split_sql_statements(sql_cmd)
        input_cmd = "\n".join(inputs)
        out, err = proc.communicate(input=input_cmd, timeout=timeout)
        logger.info(f"out:{out},err:{err}")
        if PexpectResult.FILE_PATH[3] in str(out):
            logger.info(f"no rows selected")
            return []
        result = parse_html_result(out)
        if len(result) == 0:
            return []
        logger.info(f"result:{result},pid:{pid}")
        if not parse_func:
            return result
        return [parse_func(db) for db in result]
    except Exception as error:
        logger.error(f'exec cmd {sql_cmd} failed. {str(error)}')
        raise ErrCodeException(ErrorCode.ORACLE_EXEC_SQL_FAILED.value, [sql_cmd], "execute sql error")


def exec_select_cmd_with_child(child, sql_cmd: str,  # 执行的查询语句
                               parse_func: Optional[Callable[[dict], Any]] = None):
    try:
        for setting in SELECT_SETTING:
            child.sendline(setting)
        return_code, std_out = exec_sql_cmd(child, sql_cmd, PexpectResult.FILE_PATH)
        logger.info(f"return_code:{return_code}, std_out:{std_out}")
    except Exception as exception:
        logger.error(f'exec cmd {sql_cmd} failed. {str(exception)}')
        raise ErrCodeException(ErrorCode.ORACLE_EXEC_SQL_FAILED.value, [sql_cmd], "execute sql error")
    if PexpectResult.FILE_PATH[3] in str(std_out):
        logger.info(f"query result is empty:{sql_cmd}")
        return []
    result = parse_html_result(std_out)
    if len(result) == 0:
        return []
    logger.info(f"result:{result}")
    if not parse_func:
        return result
    return [parse_func(db) for db in result]


def convert_keys_to_lower(data):
    if isinstance(data, dict):
        return {k.lower(): convert_keys_to_lower(v) for k, v in data.items()}
    elif isinstance(data, list):
        return [convert_keys_to_lower(item) for item in data]
    else:
        return data


def check_cdb_is_open(pid, login_params):
    sql_cmd = "SELECT * FROM v$database;"
    login_params.update({"pid": pid})
    result = exec_select_cmd(login_params, sql_cmd, parse_func=lambda item: item.get('CDB'))
    if not result:
        return False
    return result[0] == "YES"


def check_database_open_mode(pid, login_params):
    logger.info("Start check_database_open_mode")
    sql_cmd = "select open_mode from v$database;"
    result = exec_select_cmd(login_params, sql_cmd, parse_func=lambda item: item.get('OPEN_MODE'))
    if not result:
        return False
    return result[0]


def get_db_info(params: dict, include_cdb=False) -> DBInfo:
    logger.debug(f"get database info: pid:{params.get('pid')}")
    cmd = ("select a.log_mode, a.db_unique_name, a.database_role, a.open_mode, b.incarnation# as incarnation,"
           "to_char(a.current_scn) as current_scn,b.resetlogs_id,a.dbid as db_id "
           "from v$database a, v$database_incarnation b  where b.status='CURRENT';")
    result = exec_select_cmd(params, cmd)
    if not result:
        logger.error(f"get database info error,pid:{params.get('id')}")
        raise ErrCodeException(ErrorCode.INTERNAL_ERROR.value, message="get database info failed")
    db_info_dict = convert_keys_to_lower(result[0])
    db_info = parse_obj_as(DBInfo, db_info_dict)
    logger.info(f"get db info :{db_info}")
    instance_info = get_instance_info(params)
    db_info.status = instance_info.status
    db_info.inst_name = instance_info.instance_name
    if not include_cdb:
        return db_info
    if instance_info.is_support_cdb:
        cdb_ret = exec_select_cmd(params, "select cdb from v$database;", parse_func=lambda item: item.get('CDB'))
        db_info.cdb = cdb_ret[0]
    logger.info(f"get db info include cdb info:{db_info}")
    return db_info


def get_datafile_set(params: dict) -> set:
    logger.debug(f"get datafile set: pid:{params.get('pid')}")
    cmd = f"select file# from v$datafile;"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get("FILE#"))
    if not result:
        return set()
    return set(result)


def get_instance_info(params: dict):
    logger.debug(f"get instance info: pid:{params.get('pid')}")
    cmd = "select instance_name,status,version from v$instance;"
    result = exec_select_cmd(params, cmd)
    if not result:
        logger.error(f"get database info error,pid:{params.get('id')}")
        raise ErrCodeException(ErrorCode.INTERNAL_ERROR.value, message="get instance info failed")
    ist_info_dict = convert_keys_to_lower(result[0])
    logger.info(f"ist_info_dict:{ist_info_dict}")
    return parse_obj_as(InstanceInfo, ist_info_dict)


def get_datafile_mini_scn(params: dict):
    logger.debug(f"get instance info: pid:{params.get('pid')}")
    cmd = "select to_char(max(checkpoint_change#)) as scn from v$datafile;"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('SCN'))
    return result[0]


def list_table_space(params: dict, pdb_name=""):
    logger.debug(f"list table space info: pid:{params.get('pid')}")
    cmd = (f"select tablespace_name from dba_tablespaces where tablespace_name not in "
           f"('SYSAUX','SYSTEM','TEMP','UNDOTBS1','USERS','EXAMPLE', 'UNDOTBS2');")
    if pdb_name:
        cmd = f"alter session set container={pdb_name};\n" + cmd
    table_spaces = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('TABLESPACE_NAME'))
    return [item for item in table_spaces if item]


def get_disk_group_set(params: dict):
    cmd = "select name from v$asm_diskgroup;"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('NAME'))
    return set(result)


def get_data_node_list(params: dict):
    db_info = get_db_info(params)
    if db_info.is_primary:
        sql_cmd = f"select thread# from gv$instance where status='OPEN';"
    else:
        sql_cmd = f"select distinct thread# from v$archived_log;"
    result = exec_select_cmd(params, sql_cmd, parse_func=lambda item: item.get('THREAD#'))
    return set(result)


def get_online_redo_log_info(params: dict):
    sql_cmd = "select group#, member from v$logfile order by group#;"
    result = exec_select_cmd(params, sql_cmd, parse_func=lambda item: f"{item.get('GROUP#')};{item.get('MEMBER')}")
    logger.info(f"online redo log:{result}")
    return '\n'.join(result)


def get_pdb_list(params: dict, include_root: bool = False) -> list[PDBInfo]:
    include_root_sql = "" if include_root else "where name <> 'PDB$SEED'"
    sql_cmd = f"select name,open_mode from v$pdbs {include_root_sql};"
    result = exec_select_cmd(params, sql_cmd)
    logger.info(f"pdbs:{result}")
    return [parse_obj_as(PDBInfo, convert_keys_to_lower(item)) for item in result]


def list_pdbs(params: dict):
    sql_cmd = f"select name,open_mode from v$pdbs where open_mode = 'READ WRITE';"
    result = exec_select_cmd(params, sql_cmd)
    logger.info(f"pdbs:{result}")
    return result


def get_pdb_open_cluster_inst(params: dict):
    cmd = "select p.inst_id, i.instance_name,i.status,p.name,p.open_mode from gv$pdbs p join gv$instance i " \
          "on p.inst_id = i.inst_id where p.open_mode='READ WRITE' and p.name <> 'PDB$SEED' and i.status='OPEN' " \
          "order by p.inst_id;"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('INSTANCE_NAME'))
    return result[0] if result else ""


def get_block_change_tracking(params: dict):
    sql_cmd = "select status from v$block_change_tracking;"
    result = exec_select_cmd(params, sql_cmd, parse_func=lambda item: item.get('STATUS'))
    if not result:
        return False
    return result[0] == "ENABLED"


def get_temporary_objects(params: dict, pdb_name=""):
    cmd = f"select object_name from all_objects where temporary = 'Y';"
    if pdb_name:
        cmd = f"alter session set container={pdb_name};\n" + cmd
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('OBJECT_NAME'))
    return result


def get_table_foreign_key_dict(params: dict, pdb_name=""):
    cmd = "select table_name,constraint_name,constraint_type from dba_constraints where constraint_type='R';"
    if pdb_name:
        cmd = f"alter session set container={pdb_name};\n" + cmd
    result = exec_select_cmd(params, cmd)
    return {item.get('TABLE_NAME', ''): item for item in result}


def get_last_archive_log_next_change(params: dict, end_scn):
    cmd = f"select to_char(min(maxchange)) as change from (select max(next_change#) maxchange from v$archived_log " \
          f"where name is not null and name not like '/MNT/DATABACKUP/%' " \
          f"and next_change# <= {end_scn} group by thread#);"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('CHANGE'))
    return result[0] if result else ""


def get_first_scn_contains_time(params: dict, contains_time):
    cmd = f"select to_char(first_change#) as first_change# from v$archived_log " \
          f"where name is not null and name not like '/mnt/databackup/%' " \
          f"and to_date('{contains_time}', 'YYYY-MM-DD HH24:MI:SS') between first_time and next_time;"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('FIRST_CHANGE#'))
    return result[0] if result else ""


def get_recovery_file_dest_name(params: dict):
    cmd = "select NAME from v$recovery_file_dest;"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('NAME'))
    return result[0] if result else ""


def get_valid_destination(params: dict):
    cmd = "select destination from v$archive_dest where status='VALID';"
    return exec_select_cmd(params, cmd, parse_func=lambda item: item.get('DESTINATION'))


def get_archive_dest_list(params: dict):
    cmd = "select dest_name, status, destination from v$archive_dest where destination is not null and status='valid';"
    archive_dest = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('DESTINATION'))
    cmd = "select name, value from v$parameter where name = 'db_recovery_file_dest'"
    recovery_dest = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('VALUE'))
    combined_results = archive_dest + recovery_dest
    return [dest for dest in combined_results if dest.strip()]


def get_parameter(params: dict, parameter_name: str):
    cmd = f"show parameter {parameter_name};"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('VALUE'))
    return result[0] if result else ""


def get_local_listener(params: dict):
    return get_parameter(params, 'local_listener')


def get_db_unique_name(params: dict):
    return get_parameter(params, 'db_unique_name')


def get_spfile_path(params: dict):
    return get_parameter(params, 'pfile')


def get_asm_diskstring(params: dict):
    return get_parameter(params, 'asm_diskstring')


def get_scn_and_time(params: dict):
    cmd = f"select dbms_flashback.get_system_change_number as scn, to_char(sysdate, {TIME_FORMAT}) as time from dual;"
    result = exec_select_cmd(params, cmd)
    return [parse_obj_as(SCNAndTime, convert_keys_to_lower(item)) for item in result][0]


def get_recover_datafiles(params: dict):
    cmd = "select file# from v$datafile_header;"
    return exec_select_cmd(params, cmd, parse_func=lambda item: item.get('FILE#'))


def query_backup_copy_size(params: dict):
    cmd = '''col size_kb format 999999999999;
             select p.size_kb from(select ROUND(OUTPUT_BYTES/1024,0) SIZE_KB from v$rman_backup_job_details
             order by SESSION_KEY desc)p where rownum = 1;'''
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('SIZE_KB'))
    return result[0] if result else ""


def get_pdb_datafile_name(params: dict, pdb_name):
    cmd = f"select name from v$datafile where con_id = (select con_id from v$containers where name='{pdb_name}');"
    return exec_select_cmd(params, cmd, parse_func=lambda item: item.get('NAME'))


def get_pdb_tempfile_name(params: dict, pdb_name):
    cmd = f"select name from v$tempfile WHERE CON_ID = (SELECT CON_ID FROM V$CONTAINERS WHERE NAME='{pdb_name}');"
    return exec_select_cmd(params, cmd, parse_func=lambda item: item.get('NAME'))


def get_pdb_data_files(params: dict, pdb_names: [str]) -> [DBFile]:
    def convert(item) -> DBFile:
        return DBFile(con_id=item.get('CON_ID'), ts_name=item.get('TSNAME'), file_no=item.get('FNO'),
                      ts_file=item.get('TSFILE'))

    cmd = ("select c.con_id, c.tablespace_name tsname, f.file# fno, f.name tsfile, f.bytes/1024/1024 fmbsize "
           "from cdb_data_files c, v$datafile f where c.relative_fno = f.file# and c.con_id = 1;")
    result = exec_select_cmd(params, cmd, parse_func=convert)
    pdb_names_str = '(' + ','.join([f"'{pdb_name}'" for pdb_name in pdb_names]) + ')'
    sql = f"select t.con_id, t.name tsname, f.file# fno, f.name tsfile, f.bytes/1024/1024 fmbsize " \
          f"from v$tablespace t, v$datafile f where t.ts# = f.ts# and t.con_id = f.con_id and f.con_id in " \
          f"(select con_id from v$containers where name in {pdb_names_str}) order by bytes desc;"
    pdb_datafiles = exec_select_cmd(params, sql, parse_func=convert)
    return result + pdb_datafiles


def get_users(params: dict, pdb_name=""):
    cmd = ""
    if pdb_name:
        cmd = f"alter session set container={pdb_name};\n"
    # 过滤系统用户
    cmd += (f"select username from dba_users where username not in ('ANONYMOUS','CTXSYS','DBSNMP',"
            f"'EXFSYS','LBACSYS','MDSYS','MGMT_VIEW','OLAPSYS','OWBSYS','ORDPLUGINS','ORDSYS','OUTLN',"
            f"'SI_INFORMTN_SCHEMA','SYS','SYSMAN','SYSTEM','TSMSYS','WK_TEST','WKSYS','WKPROXY','WMSYS',"
            f"'XDB','APEX_PUBLIC_USER','DIP','FLOWS_30000','FLOWS_FILES','MDDATA','ORACLE_OCM',"
            f"'SPATIAL_CSW_ADMIN_USR','SPATIAL_WFS_ADMIN_USR','XS$NULL','APPQOSSYS','DBSFWUSER','GGSYS',"
            f"'DVSYS','DVF','GSMADMIN_INTERNAL','GSMCATUSER','SYSBACKUP','REMOTE_SCHEDULER_AGENT','GSMUSER',"
            f"'SYSRAC','OJVMSYS','AUDSYS','SYSKM','ORDDATA','SYS$UMF','SYSDG');")
    users = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('USERNAME'))
    return [item for item in users if item]


def list_table_by_owner(params: dict, table_param_list: ListTableByOwnerParam):
    logger.info(f"Start list_table_by_owner")
    cmd = ""
    if table_param_list.pdb_name:
        cmd = f"alter session set container={table_param_list.pdb_name};\n"
    begin = (table_param_list.page - 1) * table_param_list.page_size + 1
    end = table_param_list.page * table_param_list.page_size
    cmd += (f"select * from (select ROWNUM r, t.OWNER, t.TABLE_NAME, t.TABLESPACE_NAME from dba_tables t "
            f"where OWNER = '{table_param_list.owner}' and OWNER != 'SYS') where r between {begin} and {end};")
    result = exec_select_cmd(params, cmd)
    return result


def get_omf_dest(params: dict):
    cmd = "select value from v$parameter where name='db_create_file_dest';"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('VALUE'))
    return result[0] if result else ''


def get_pdb_info(params: dict, pdb_name):
    cmd = f"select name, guid from v$pdbs where name = '{pdb_name}';"
    result = exec_select_cmd(params, cmd)
    if not result:
        return "", ""
    return result[0].get("GUID"), result[0].get("NAME")


def get_min_of_last_log_scn_by_thread(params: dict, log_is_backed_up):
    logger.info(f"min of last log scn by thread, pid:{params.get('id')}")
    node_list = get_database_node_list(params)
    if not node_list:
        logger.error(f"get node list failed.")
        raise ErrCodeException(ErrorCode.INTERNAL_ERROR.value, message="archive thread# is empty")
    scn_min = "300000000000000"
    last_log_time = ""
    min_scn_log_info = []
    for node in node_list:
        last_log_scn = get_last_log_scn(params, node, log_is_backed_up)
        if not last_log_scn:
            logger.warn(f"get last log scn empty")
            continue
        logger.info(f"Last log scn is {last_log_scn}, scn_min:{scn_min}.")
        if int(last_log_scn) < int(scn_min):
            scn_min = last_log_scn
            last_log_time = get_log_time(params, node, log_is_backed_up, scn_min)
    if int(scn_min) < 300000000000000:
        min_scn_log_info.append(scn_min)
        min_scn_log_info.append(last_log_time)
    logger.info(f"Get min of last log scn by thread success, min_scn_log_info:{min_scn_log_info}")
    return min_scn_log_info


def get_database_node_list(params):
    db_info = get_db_info(params)
    if db_info.is_primary:
        logger.info(f"Get database node list primary.")
        sql_cmd = f"select thread# from gv$instance where status='OPEN';"
    else:
        logger.info(f"Get database node list standby.")
        sql_cmd = f"select distinct thread# from v$archived_log;"
    result = exec_select_cmd(params, sql_cmd, parse_func=lambda item: item.get('THREAD#'))
    logger.info(f"get_database_node_list:{result}")
    return result


def get_last_log_scn(params, thread, log_is_backed_up):
    logger.info(f"get last log scn.")
    thread_sql = f"and THREAD#={thread}" if thread else ""
    sql = f"select to_char(max(next_change#)) from v$archived_log where {log_is_backed_up} {thread_sql};"
    result = exec_select_cmd(params, sql, parse_func=lambda item: item.get('TO_CHAR(MAX(NEXT_CHANGE#))'))
    if not result:
        logger.error(f"Failed to get last archive log scn, cmd: {sql}.")
        return ""
    return result[0]


def get_log_time(params: dict, thread: str, log_is_backed_up: str, from_scn: str):
    logger.info(f"get log time.")
    thread_sql = f"and THREAD#={thread}" if thread else ""
    sql = f"select to_char(next_time,{TIME_FORMAT}) as next_time from v$archived_log where next_change#={from_scn}" \
          f" and {log_is_backed_up} {thread_sql};"
    result = exec_select_cmd(params, sql, parse_func=lambda item: item.get('NEXT_TIME'))
    logger.info(f"get_log_time,result:{result}")
    return result[0] if result else ""


def get_all_table_space_from_pdb(params: dict, pdb_name):
    logger.info(f"get_all_table_space_from_pdb, pdb_name:{pdb_name},pid:{params.get('pid')}")
    cmd = f"select name from v$tablespace where con_id = (select con_id from v$containers where name = '{pdb_name}');"
    result = exec_select_cmd(params, cmd, parse_func=lambda item: item.get('NAME'))
    return set(result)
